package com.wpre.vampiremq.mqserver.core;

import com.wpre.vampiremq.common.Consumer;
import com.wpre.vampiremq.common.ConsumerEnv;
import com.wpre.vampiremq.common.MQException;
import com.wpre.vampiremq.mqserver.VirtualHost;
import org.springframework.cglib.core.Block;

import java.util.concurrent.*;

/**
 * Created with IntelliJ IDEA.
 * Description:实现消费消息的核心逻辑
 * User: LianBao
 * Date: 2025-01-18
 * Time: 16:36
 */
public class ConsumerManager {
    private VirtualHost virtualHost;//用于操作数据

    //线程池,用于执行具体的回调任务
    private ExecutorService wokePoll = Executors.newFixedThreadPool(4);

    //令牌队列,每次拿到一个令牌,才从对应的队列中取一个消息
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();

    //扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost virtualHost) {
        this.virtualHost = virtualHost;
        scannerThread = new Thread(() -> {
            //扫描线程要完成的工作
            while (true) {
                try {
                    //1.拿到令牌
                    String queueName = tokenQueue.take();
                    //2.根据令牌找到队列
                    MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MQException("根据令牌取队列,发现队列不存在");
                    }
                    //3.从队列中消费消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }

                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        scannerThread.setDaemon(true);//设置为后台线程
        scannerThread.start();

    }

    //通知 去进行消费
    public void notifyConsume(String queueName) throws InterruptedException {
        //把队列名加到令牌队列中
        tokenQueue.put(queueName);
    }


    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        //1.找到对应的队列
        MSGQueue queue = virtualHost.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MQException("队列不存在");
        }
        //2.构造ConsumerEnv对象
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, consumer, queueName, autoAck);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);//添加一个消费者

            int messageCount = virtualHost.getMemoryDataCenter().getMessageCount(queueName);
            //查看有多少消息,如果有消息就先消费
            for (int i = 0; i < messageCount; i++) {
                //调用一次就消费一条消息
                consumeMessage(queue);
            }

        }


    }

    //消费消息
    private void consumeMessage(MSGQueue queue) {
        //1.按照轮询方式,找个消费者来,进行消费
        ConsumerEnv consumerEnv = queue.chooseConsumer();
        if (consumerEnv == null) {
            //当前队列没有消费者(订阅者)
            return;
        }
        //2.从队列中取出一个消息
        Message message = virtualHost.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            //当前队列没有消息
            return;
        }

        //3.把消息带人回调函数,让线程池执行
        wokePoll.submit(() -> {
            try {
                //为了确保消息不丢失,先把消息放入待确认消息的集合中
                virtualHost.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                //执行回调操作
                consumerEnv.getConsumer().handleDelivery(consumerEnv.getConsumerTag(), message.getBasicProperties(), message.getBody());
                //如果是自动应答,直接删除消息
                //如果是手动应答,后续等消费者调用basicAck后删除
                if (consumerEnv.isAutoAck()) {
                    //删除硬盘消息
                    if (message.getDeliverMode() == 2) {
                        virtualHost.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    //删除待确认集合消息
                    virtualHost.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    //删除内存中的消息
                    virtualHost.getMemoryDataCenter().removeMessage(message.getMessageId());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });


    }
}
